package com.ai.nsa.inputer;

import com.ai.nsa.NSACache;
import com.ai.nsa.NSASerial;
import com.ai.nsa.utils.MapTools;
import com.ai.nsa.utils.hbase.HBaseOperations;
import com.ai.nsa.utils.hbase.model.HBaseRow;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.Map;

import static com.ai.nsa.NSAConstants.*;
import static com.ai.nsa.NSAFields.SLAVE_MKT;

/**
 * 营销业务办理
 *
 * @author Feng Chen 2017/6/19.
 */
@Component(value = "mktInputer")
public class TradeMktInputer extends AbstractInputer {

    @Autowired
    public TradeMktInputer(NSASerial serial, HBaseOperations hbase, NSACache cache) {
        super(serial, hbase, cache);
    }

    @Override
    public void rename(Map<String, Object> row) {
        row.put(COLUMN_TRADE_DATE, convert(MapTools.getString(row, COLUMN_ACCEPT_DATE)));
    }

    @Override
    public void putSlave(String rowKey, Map<String, Object> row) throws Exception {
        HBaseRow hBaseRow = new Converter() {

            @Override
            public HBaseRow convert(Map<String, Object> event) {
                if (event == null || event.isEmpty()) {
                    return null;
                }

                HBaseRow row = new HBaseRow();

                // 设置 列组
                for (String key : event.keySet()) {

                    if (event.get(key) == null || !SLAVE_MKT.contains(key)) {
                        continue;
                    }

                    row.addValue(HBASE_FAMILY, key, MapTools.getString(event, key));
                }

                return row;
            }

        }.convert(row);
        hBaseRow.setRowKey(rowKey);
        hbase.put(HBASE_SLAVE_TABLE_MKT + UNDERLINE + tableRoute.getTable(rowKey), hBaseRow);
    }

}
